package com.cmb.lk50.common;

import com.cmb.lk50.common.extractor.BodyParamExtractor;
import com.cmb.lk50.common.extractor.QueryParamExtractor;
import com.cmb.lk50.common.receiver.HttpStatusReceiver;
import com.cmb.lk50.common.receiver.ResponseBodyReceiver;
import com.cmb.lk50.common.receiver.ResponseReceiver;
import com.cmb.lk50.pojo.AbstractLog;
import com.cmb.lk50.pojo.AccessInfo;
import com.cmb.lk50.sink.LogSink;
import com.cmb.lk50.sink.kafka.KafkaKeyGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.web.method.HandlerMethod;
import org.springframework.web.reactive.result.method.annotation.RequestMappingHandlerMapping;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.WebFilter;
import org.springframework.web.server.WebFilterChain;
import reactor.core.publisher.Mono;

import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Map;


/**
 * Webflux环境下的日志拦截主流程
 * <p>
 * Author       : 蒋俊钊 80256973
 * Date         : 2020/7/17
 * Copyright    (C) ChinaMerchantsBank
 */
public class WebFluxAccessFilter extends Sinkable implements WebFilter {

    private static final Logger LOGGER = LoggerFactory.getLogger(WebFluxAccessFilter.class);

    private static HttpStatusReceiver httpStatusReceiver = new HttpStatusReceiver();

    private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    @Resource
    private RequestMappingHandlerMapping requestMappingHandlerMapping;

    @Autowired(required = false)
    private List<LogSink> logSinks;

    @Resource
    private Map<Class, QueryParamExtractor> cacheExtractor;

    @Resource
    private Map<Class, ResponseReceiver> cacheReceiver;

    @Resource
    private Map<Class, KafkaKeyGenerator<AbstractLog>> cacheKeyer;

    /**
     * 作用同AccessInterceptor
     */
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        HandlerMethod handlerMethod = (HandlerMethod) requestMappingHandlerMapping.getHandler(exchange).toProcessor().peek();
        if (handlerMethod == null) {
            return chain.filter(exchange);
        }
        AccessLog annotation = handlerMethod.getMethod().getAnnotation(AccessLog.class);
        if (annotation == null) {
            return chain.filter(exchange);
        }

        try {
            ServerHttpRequest request = exchange.getRequest();
            ServerHttpResponse response = exchange.getResponse();

            AccessInfo accessInfo = new AccessInfo();
            InetSocketAddress localAddress = request.getLocalAddress();
            if (localAddress != null) {
                accessInfo.setHostName(localAddress.getHostName());
                accessInfo.setHostIp(localAddress.getAddress() != null ? localAddress.getAddress().getHostAddress(): "");
                accessInfo.setHostPort(localAddress.getPort());
            }

            accessInfo.setMethod(request.getMethodValue());
            accessInfo.setPath(request.getPath().value());

            long startTime = System.currentTimeMillis();

            InetSocketAddress remoteAddress = request.getRemoteAddress();
            if (remoteAddress != null) {
                String invokeIP;
                List<String> forwardIP = request.getHeaders().get("x-forwarded-for");
                if (forwardIP == null || forwardIP.isEmpty()) {
                    invokeIP = remoteAddress.getAddress().toString();
                } else {
                    String ip = forwardIP.get(0);
                    invokeIP = ip.contains(",") ? ip.split(",")[0] : ip;
                }
                accessInfo.setInvokeIp(invokeIP);
                accessInfo.setInvokePort(remoteAddress.getPort());
            }

            // wrapper
            ServerHttpRequest wrapperRequest = new BodyParamExtractor().new WrapperServerHttpRequestDecorator(request);
            ServerHttpResponse wrapperResponse = new ResponseBodyReceiver().new WrapperServerHttpResponseDecorator(response);

            ServerWebExchange webExchange = exchange.mutate().request(wrapperRequest).response(wrapperResponse).build();
            return chain.filter(webExchange)
              .doAfterTerminate(() -> {
                  QueryParamExtractor extractor = cacheExtractor.get(annotation.extractor());
                  ResponseReceiver receiver = cacheReceiver.get(annotation.receiver());

                  // 提取入参
                  if (extractor != null) {
                      String queryParam = extractor.extract(wrapperRequest);
                      accessInfo.setQueryParam(queryParam);
                  }

                  ServerHttpResponse httpResponse = webExchange.getResponse();

                  // 提取结果
                  if (receiver != null && !(receiver instanceof HttpStatusReceiver)) {
                      accessInfo.setResponseContent(receiver.receive(httpResponse));
                  }
                  accessInfo.setStatus(httpStatusReceiver.receive(httpResponse));

                  accessInfo.setCostTime(System.currentTimeMillis() - startTime);
                  accessInfo.setLogTime(DATE_TIME_FORMATTER.format(LocalDateTime.now()));


                  AbstractLog log = null;
                  Class<? extends AbstractLog> logClazz = annotation.value();
                  if (logClazz != AccessInfo.class) {
                      try {
                          log = logClazz.newInstance().convert(accessInfo);
                      } catch (InstantiationException | IllegalAccessException e) {
                          LOGGER.error("转换AccessInfo异常, [{}]", e.getMessage());
                      }
                  } else {
                      log = accessInfo;
                  }

                  if (log != null) {
                      doSinks(logSinks, log, cacheKeyer.get(annotation.kafkaKeyer()));
                  }
              });
        } catch (RuntimeException e) {
            LOGGER.warn(e.getMessage(), e);
            return chain.filter(exchange);
        }
    }

}
